package function;

import bean.createTableBean;
import org.apache.commons.compress.utils.CharsetNames;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class createTableUtils {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //读取数据
        DataStreamSource<String> dataStreamSource = env.readTextFile("C:\\Users\\Administrator\\Desktop\\data.txt", CharsetNames.UTF_8);
        //dataStreamSource.print();

        //处理数据拼接sql
        SingleOutputStreamOperator<String> streamOperator = dataStreamSource.map(line -> {
            String[] split = line.split("#");
            createTableBean createTableBean = new createTableBean();
            createTableBean.setTableName(split[0]);
            createTableBean.setColName(split[1]);
            createTableBean.setType(split[2]);
            createTableBean.setComment(split[3]);
            return createTableBean;
        }).keyBy(v -> v.getTableName()).map(new richTableMapFunction());

        //streamOperator.print();

        env.execute("createTableUtils");
    }
}
